package org.finesys.common.hazelcast.core.support;

import com.hazelcast.client.HazelcastClient;
import com.hazelcast.client.config.ClientConfig;
import com.hazelcast.client.config.ClientConnectionStrategyConfig;
import com.hazelcast.client.config.ClientNetworkConfig;
import com.hazelcast.core.HazelcastInstance;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;

import java.util.Objects;
import java.util.StringJoiner;

@RequiredArgsConstructor
@Slf4j
public class HazelcastClientFactory implements FactoryBean<HazelcastInstance>, DisposableBean {

    private final HazelcastProperties hazelcastProperties;
    private final HazelcastServerInstance hazelcastServerInstance;

    private HazelcastInstance instance;

    public HazelcastInstance getInstance() {
        if (Objects.nonNull(hazelcastServerInstance)) {
            log.info("本地启动Hazelcast节点:{}", hazelcastServerInstance);
        }
        if (Objects.nonNull(instance)) {
            return instance;
        }

        ClientConfig clientConfig = new ClientConfig();
        // 配置集群名称
        clientConfig.setClusterName(hazelcastProperties.getClusterName());
        // 配置实例名称
        clientConfig.setInstanceName(hazelcastProperties.getInstanceName());
        // 配置网络
        configNetwork(clientConfig.getNetworkConfig());
        // 配置连接模式
        configConnectionStrategy(clientConfig.getConnectionStrategyConfig());
        // 创建实例
        this.instance = HazelcastClient.newHazelcastClient(clientConfig);

        return instance;
    }

    private void configConnectionStrategy(ClientConnectionStrategyConfig connectionStrategyConfig) {
        connectionStrategyConfig.setAsyncStart(false).setReconnectMode(ClientConnectionStrategyConfig.ReconnectMode.ASYNC);
    }

    private void configNetwork(ClientNetworkConfig networkConfig) {
        networkConfig.addAddress(members()).setSmartRouting(true).setConnectionTimeout(5000);
    }

    private String[] members() {
        StringJoiner sj = new StringJoiner(",");
        hazelcastProperties.getMembers().stream().flatMap(m -> hazelcastProperties.getPort().stream().map(p -> m + ":" + p)).forEach(sj::add);
        return sj.toString().split(",");
    }

    @Override
    public void destroy() throws Exception {
        instance.shutdown();
    }


    @Override
    public HazelcastInstance getObject() throws Exception {
        return getInstance();
    }

    @Override
    public Class<?> getObjectType() {
        return HazelcastInstance.class;
    }
}
